package net.xuele.learn.flink;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 下面的程序从一系列整数中连续减去 1，直到它们达到零：
 */
public class IterativeStreamTest {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Long> someIntegers = env.generateSequence(0, 1000);
        IterativeStream<Long> iteration = someIntegers.iterate();
        DataStream<Long> minusOne = iteration.map((MapFunction<Long, Long>) value -> value - 1);

        DataStream<Long> stillGreaterThanZero = minusOne.filter((FilterFunction<Long>) value -> (value > 0));

        // 关闭迭代并定义迭代尾部
        // 一种常见的模式是使用过滤器将反馈的流部分和向前传播的流部分分开。
        // 例如，这些过滤器可以定义“终止”逻辑，其中允许元素向下游传播而不是被反馈
        iteration.closeWith(stillGreaterThanZero);

        DataStream<Long> lessThanZero = minusOne.filter((FilterFunction<Long>) value -> (value <= 0));
    }
}
